import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousProcessingTimeTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import redis.clients.jedis.Jedis;

import java.util.Calendar;
import java.util.Map;
import java.util.Properties;

/**
 * @author : yujie
 * @date : 2021/12/8
 */
public class fk {
    public static JRedisUtil conn = new JRedisUtil("127.0.0.1",6379);
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置并行度1
        env.setParallelism(1);

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.234.141:9092");
        // 下面这些次要参数
        properties.setProperty("group.id", "group");
        properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("auto.offset.reset", "latest");

//        Calendar c = Calendar.getInstance();
//        int hour1 = c.get(Calendar.HOUR_OF_DAY);
//        int minute1 = c.get(Calendar.MINUTE);
//        int second1 = c.get(Calendar.SECOND);
//        int t1=hour1*3600+minute1*60+second1;
//        final int[] rsum = {0};
        DataStream<String> dataStream = env.addSource(new FlinkKafkaConsumer<String>("example", new SimpleStringSchema(),properties));

        //读取ordeiid数据
        DataStream<String> filterStream = dataStream.filter(new FilterFunction<String>() {

            public boolean filter(String value) throws Exception {
                return value.startsWith("{\"data\":[{\"order_id\"");
            }
        });

        DataStream<String> dataStream1=filterStream.map(line->{
            String jsonObj = JSON.parseObject(line).getString("data");
            String r1 = jsonObj.replace("]","");
            String r2 = r1.replace("[","");
            JSONObject jsonObj1 = JSON.parseObject(r2);
            //书的前十
            int book_num = Integer.parseInt(jsonObj1.getString("book_num"));
            String book_name = jsonObj1.getString("book_name");
            conn.zincrby("book10",book_num,book_name);
            //出版社前十
            String book_press = jsonObj1.getString("book_press");
            conn.zincrby("press10",book_num,book_press);
            String booktype = jsonObj1.getString("book_type");
            conn.zincrby("brank",1,booktype);
            //rsum[0] = rsum[0] +1;
            return "";
        });
        //读取orderdetailid数据
        DataStream<String> filterStream2 = dataStream.filter(new FilterFunction<String>() {

            public boolean filter(String value) throws Exception {
                return value.startsWith("{\"data\":[{\"orderdetailid\"");
            }
        });

        DataStream<String> dataStream2 = filterStream2
                .map(line->{
                    String jsonObj1 = JSON.parseObject(line).getString("data");
                    String r1 = jsonObj1.replace("]","");
                    String r2 = r1.replace("[","");
                    JSONObject jsonObj = JSON.parseObject(r2);
                    Float total=Float.parseFloat(jsonObj.getString("total").substring(0,jsonObj.getString("total").indexOf(".")+2));
                    conn.incrBy("total",total);
                    conn.incrBy("order_sum",1);
                    String re = jsonObj.getString("recipient");
                    conn.sadd("rep",re);
                    String place = jsonObj.getString("receiveplace").substring(0,2);
                    conn.zincrby("place",1,place);

                    //rsum[0] = rsum[0] +1;
                    return "";
                });
//        int hour2 = c.get(Calendar.HOUR_OF_DAY);
//        int minute2 = c.get(Calendar.MINUTE);
//        int second2 = c.get(Calendar.SECOND);
//        int t2=hour2*3600+minute2*60+second2;
//        if((t2-t1)!=0){
//            int speed=(rsum[0]/(t2-t1));
//            System.out.println(speed);
//            rsum[0]=0;
//            t1=t2;
//        }
        dataStream2.print();

        env.execute();
    }
}